package util;

import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.KV;
import org.joda.time.DateTime;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by emily on 17-12-13.
 */
public class MyTimestampFunction implements SerializableFunction<KV<String, String>, Instant> {

    private static final Logger LOG = LoggerFactory.getLogger(MyTimestampFunction.class);
    private long firstTimestamp = Long.MIN_VALUE;


    public Instant apply(KV<String, String> input){

        String[] temps = input.getValue().split(",");

        if(firstTimestamp == Long.MIN_VALUE){
            firstTimestamp = Long.valueOf(temps[1]);
            System.out.println("first time: " + temps[1]);
        }
        Long l = Long.valueOf(temps[1])-firstTimestamp;
        DateTime t = new DateTime(l);

        //System.out.println("MyTimestampFn: " + l);
        return t.toInstant();


    }
}
